package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.bridge.scala._

object TopNTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    val click_table = tEnv.fromDataStream(clickStream,
      $("user"),
      $("url"),
      $("ts").rowtime())
    tEnv.createTemporaryView("click_table", click_table)

    // 统计url top n
    val subSql =
      """
        |select window_start, window_end, url, count(distinct user) uv
        |from table(tumble(table click_table, descriptor(ts), interval '10' second))
        |group by window_start, window_end, url
        |""".stripMargin

    val sql =
      s"""
         |select *
         |from (
         |    select *,
         |        row_number() over(partition by window_start, window_end order by uv desc) as row_num
         |    from (${subSql})
         |)
         |where row_num <= 3
         |""".stripMargin


    tEnv.sqlQuery(sql).execute().print()

    env.execute("TopNTest")
  }
}
